package com.bdqn.qnnews.schedule.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.json.JSONUtil;
import com.bdqn.qnnews.common.constants.RedisConstants;
import com.bdqn.qnnews.common.constants.ScheduleConstants;
import com.bdqn.qnnews.schedule.api.dto.TaskDto;
import com.bdqn.qnnews.schedule.convert.TaskConvert;
import com.bdqn.qnnews.schedule.entity.Taskinfo;
import com.bdqn.qnnews.schedule.entity.TaskinfoLogs;
import com.bdqn.qnnews.schedule.mapper.TaskinfoLogsMapper;
import com.bdqn.qnnews.schedule.mapper.TaskinfoMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * @author xlzhang
 * @date 2024/1/16 9:36
 */
@Slf4j
@RequiredArgsConstructor
@Service
public class TaskServiceImpl implements com.bdqn.qnnews.schedule.service.TaskService {
    private final TaskinfoMapper taskinfoMapper;
    private final TaskinfoLogsMapper taskinfoLogsMapper;
    private final StringRedisTemplate redisTemplate;

    // 开启本地事务
    @Transactional(rollbackFor = Exception.class)
    @Override
    public Long addTask(TaskDto taskDto) {
        // 添加任务到数据库 DB
        // 因为当前涉及到两种数据库，如果中途发生异常，事务控制不能控制redis，
        // 因此这里要先保存数据到mysql，再保存到redis中,这样可以尽量保证数据存储的一致性
        // 1. 先把任务添加到DB，也就是mysql数据库
        addTaskToDb(taskDto);

        // 判断当前时间是否 <= 预设时间，如果是就需要将对应的任务时间缓存到redis队列中
        // 注意，能到这里的任务都是延时任务，因为既时任务已经在发布文章的方法中已经分流了
        // 2. 再把任务添加Redis
        addTaskToCache(taskDto);

        // 返回任务id
        return taskDto.getTaskId();
    }

    @Override
    public List<TaskDto> pollTask(Integer taskTopic) {
        // 拼接要查询数据的key
        String redisKey = RedisConstants.TASK_TOPIC_PREFIX + taskTopic;
        // 直接从redis查询数据
        final Set<String> taskList = redisTemplate.opsForZSet().rangeByScore(redisKey, 0, System.currentTimeMillis());
        // 创建一个保存任务对象的列表
        final ArrayList<TaskDto> taskDtoList = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(taskList)) {
            // 遍历并转换数据
            for (String taskJson : taskList) {
                // 将json数据转换为对象
                final TaskDto taskDto = JSONUtil.toBean(taskJson, TaskDto.class);
                // 消费任务
                // 第一步应该是先根据获取到的任务信息 发布文章到app端，发布成功后再执行后续代码

                // 先从redis中移除已经发布成功的消息数据
                redisTemplate.opsForZSet().remove(redisKey,taskJson);

                //更新DB数据
                updateTaskToDB(taskDto);

                // 将任务对象保存到列表
                taskDtoList.add(taskDto);
            }
        }
        return taskDtoList;
    }

    /**
     * 更新mysql数据库中的任务信息
     * @param taskDto 任务数据
     */
    private void updateTaskToDB(TaskDto taskDto) {
        try {
            // 再从mysql数据库中移除taskInfo
            taskinfoMapper.deleteById(taskDto.getTaskId());

            // 再修改mysql数据库中taskInfoLogs的状态
            // 当前的定时任务日志表是需要实现乐观锁的，所以需要先查询到更新前的version版本
            final TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskDto.getTaskId());
            // 修改状态
            taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);
            // 更新数据
            taskinfoLogsMapper.updateById(taskinfoLogs);
        } catch (Exception e) {
            log.error("--更新任务消费日志数据异常：{}",e.getMessage());
            throw new RuntimeException(e);
        }
    }

    /**
     * 保存数据到redis缓存队列中
     * @param taskDto 任务数据
     */
    private void addTaskToCache(TaskDto taskDto) {
        // 计算获取5分钟后的时间
        final LocalDateTime executeTime = LocalDateTimeUtil.of(taskDto.getExecuteTime());
        // 计算预设时间，当前默认设置5分钟之后的日期
        final LocalDateTime offset = LocalDateTimeUtil.offset(LocalDateTime.now(), 5, ChronoUnit.MINUTES);
        // 比较发布日期是否在预设日期之前，如果是就代表当前需要添加到redis缓存队列
        if (executeTime.isBefore(offset)){
            // 保存到redis
            // 1. 创建一个key，用来保存数据
            String redisKey = RedisConstants.TASK_TOPIC_PREFIX + taskDto.getTaskTopic();
            /*
                2. 保存数据到redis
                参数： 1. 缓存队列的名称(key) 2. 要缓存的任务数据  3. score需要使用发布时间的毫秒值
             */
            redisTemplate.opsForZSet().add(redisKey, JSONUtil.toJsonStr(taskDto), taskDto.getExecuteTime());
        }
    }

    // 保存定时任务到mysql数据库中
    private void addTaskToDb(TaskDto taskDto) {
        try {
            // 先保存数据到taskinfo
            final Taskinfo taskinfo = TaskConvert.INSTANCE.dtoToEntity(taskDto);
            // 当前的发布时间是被忽略的，需要手动设置
            taskinfo.setExecuteTime(LocalDateTimeUtil.of(taskDto.getExecuteTime()));
            // 调用mapper保存数据
            taskinfoMapper.insert(taskinfo);
            // 将添加数据成功后的id赋值给taskDto，用于保存数据到logs表的时候id是相同(实现主键相同来建立关系)
            // mybatisPlus在保存数据后会默认会写主键id
            taskDto.setTaskId(taskinfo.getTaskId());

            // 再保存到taskinfologs
            final TaskinfoLogs taskinfoLogs = TaskConvert.INSTANCE.infoToInfoLogs(taskinfo);
            // 设置乐观锁的版本初始值
            taskinfoLogs.setVersion(1);
            // 设置当前的状态 为初始化状态
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
            // 保存数据
            taskinfoLogsMapper.insert(taskinfoLogs);
        } catch (Exception e) {
            log.error("--定时任务保存到数据库异常：{}",e.getMessage());
            throw new RuntimeException(e);
        }
    }
}
